﻿namespace JESAI.DistributedLock.ZooKeeper;

using org.apache.zookeeper;
using org.apache.zookeeper.data;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Threading.Tasks;

internal static class ZooKeeperNodeCreator
{
    /// <summary>
    /// 请参阅https://zookeeper.apache.org/doc/r3.5.4-beta/zookeeperProgrammers.html在“内置ACL方案”下
    /// </summary>
    public static readonly ACL PublicAcl = new(0x1f, new Id("world", "anyone"));

    public static async Task<string> CreateEphemeralSequentialNode(
        this ZooKeeperConnection connection,
        ZooKeeperPath directory,
        string namePrefix,
        IEnumerable<ACL> aclEnumerable,
        bool ensureDirectoryExists)
    {
        // 如果我们负责确保目录，则此算法会循环，直到EnsureDirectory失败或遇到目录不存在以外的错误。这支持并发创建节点和创建目录，以及创建节点和删除目录。

        var acl = aclEnumerable.ToList();
        while (true)
        {
            var createdDirectories = ensureDirectoryExists
                ? await EnsureDirectoryAndGetCreatedPathsAsync().ConfigureAwait(false)
                : Array.Empty<ZooKeeperPath>();

            try
            {
                return await connection.ZooKeeper.createAsync($"{directory}{ZooKeeperPath.Separator}{namePrefix}", data: Array.Empty<byte>(), acl, CreateMode.EPHEMERAL_SEQUENTIAL).ConfigureAwait(false);
            }
            catch (KeeperException.NoNodeException ex)
            {
                // 如果我们没有确保目录，请重新显示一条更有用的错误消息。否则，接受错误并再次循环
                if (!ensureDirectoryExists)
                {
                    throw new InvalidOperationException($"Node '{directory}' does not exist", ex);
                }
            }
            catch
            {
                // 在出现未处理的错误时，清理我们创建的所有目录
                await TryCleanUpCreatedDirectoriesAsync(createdDirectories).ConfigureAwait(false);

                throw;
            }
        }

        async Task<IReadOnlyList<ZooKeeperPath>> EnsureDirectoryAndGetCreatedPathsAsync()
        {
            // 此算法会循环，直到目录存在，或者我们的创建尝试因NoNodeException或NodeExistsException以外的原因而失败。这支持并发目录创建以及通过乐观并发进行并发创建和删除

            var toCreate = new Stack<ZooKeeperPath>();
            toCreate.Push(directory);
            List<ZooKeeperPath>? created = null;
            do
            {
                var directoryToCreate = toCreate.Peek();
                if (directoryToCreate == ZooKeeperPath.Root)
                {
                    throw new InvalidOperationException($"Received {typeof(KeeperException.NoNodeException)} when creating child node of directory '{ZooKeeperPath.Root}'");
                }

                try
                {
                    await connection.ZooKeeper.createAsync(directoryToCreate.ToString(), data: Array.Empty<byte>(), acl, CreateMode.PERSISTENT).ConfigureAwait(false);
                    toCreate.Pop();
                    (created ??= new List<ZooKeeperPath>()).Add(directoryToCreate);
                }
                catch (KeeperException.NodeExistsException) // 其他人创造了它
                {
                    toCreate.Pop();
                }
                catch (KeeperException.NoNodeException) // 需要创建父级
                {
                    toCreate.Push(directoryToCreate.GetDirectory()!.Value);
                }
                catch
                {
                    // 在未处理的故障中，尝试清理我们的工作
                    if (created != null) { await TryCleanUpCreatedDirectoriesAsync(created).ConfigureAwait(false); }

                    throw;
                }
            }
            while (toCreate.Count != 0);

            return created ?? (IReadOnlyList<ZooKeeperPath>)Array.Empty<ZooKeeperPath>();
        }

        async Task TryCleanUpCreatedDirectoriesAsync(IReadOnlyList<ZooKeeperPath> createdDirectories)
        {
            try
            {
                // 按与创建相反的顺序删除
                for (var i = createdDirectories.Count - 1; i >= 0; --i)
                {
                    await connection.ZooKeeper.deleteAsync(createdDirectories[i].ToString()).ConfigureAwait(false);
                }
            }
            catch
            {
                // 吞咽错误，因为这种清理很可能会以与创建相同的方式失败
            }
        }
    }
}
